/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package nonmr;

import avroserialization.AvroSerialization;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.Calendar;
import java.util.Date;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import overfeatfeatureextraction.AvroImage;

/**
 *
 * @author mwh
 */
public class HBasePutter {

    public static void main(String args[]) throws Exception {
        HBasePutter hbasePutter = new HBasePutter();
        hbasePutter.put(null);
    }
    HTable imageTable;
    HTable queueTable;

    AvroSerialization avroByteSerializer;
    
    public HBasePutter() throws IOException {
        Configuration conf = HBaseConfiguration.create();
        imageTable = new HTable(conf, "images");
        queueTable = new HTable(conf, "imagequeue");
        avroByteSerializer = AvroSerialization.INSTANCE;
    }

    public void put(AvroImage img) throws IOException {

        
        byte[] key = img.getMd5().array();
        Get get = new Get(key);
        Result result = imageTable.get(get);
        if (result.isEmpty()) {
            System.err.println("Empty");
            
            
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            avroByteSerializer.serializeWithSchema(baos, img);
            baos.close();
            byte[] avroBytes = baos.toByteArray();
            
            Put imageTablePut = new Put(key);

            imageTablePut.add(Bytes.toBytes("d"), Bytes.toBytes("avroimage"), avroBytes);
            imageTable.put(imageTablePut);
            
            Put queueTablePut = new Put(key);

            Date d = new Date(System.currentTimeMillis());
            queueTablePut.add(Bytes.toBytes("d"), Bytes.toBytes("d"), Bytes.toBytes(d.toString()));
            queueTable.put(queueTablePut);
            
            
            
        } else {
            System.err.println("Not empty");
        }
    }
}
